-- 订单更新量较小，不采取 （1天ODS+29天DWD）方式递归计算；直接取30天ODS数据去重
with mark_expand_new as (
    select
        id
        ,order_id
        ,waybill_id
        ,order_source_code
        ,field_name
        ,field_value
        ,input_time
        ,update_time
        ,is_delete
        ,row_number() over(partition by id order by dt desc) as row_id
        ,date_format(input_time, 'yyyy-MM-dd') as dt
    from jms_ods.yl_oms_order_mark_expand
    where dt >= date_add('{{ execution_date | cst_ds }}', -30) and dt <= '{{ execution_date | cst_ds }}'
)

insert overwrite table jms_dwd.dwd_yl_oms_order_mark_expand_base_dt partition(dt)
select
    id
    ,order_id
    ,waybill_id
    ,order_source_code
    ,field_name
    ,field_value
    ,input_time
    ,update_time
    ,is_delete
    ,dt
from  mark_expand_new
where row_id = 1
and dt >= date_add('{{ execution_date | cst_ds }}', -30)
and dt <= '{{ execution_date | cst_ds }}'
distribute by dt, abs(hash(id)) % 10 ;